package com.atguigu.app;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @className: Test4_SQL_LookUp
 * @author: LinCong
 * @description:
 * @date: 2023/1/30 19:28
 * @version: 1.0
 */
public class Test4_SQL_LookUp {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String sql = "create TEMPORARY TABLE base_dic( " +
                "  `dic_code` String, " +
                "  `dic_name` String, " +
                "  `parent_code` String, " +
                "  `create_time` String, " +
                "  `operate_time` String " +
                ")WITH ( " +
                "  'connector' = 'jdbc', " +
                "  'url' = 'jdbc:mysql://hadoop3-02:3306/gmall', " +
                "  'driver' = 'com.mysql.cj.jdbc.Driver', " +
//                维表数据不变or会改变，但是对数据的准确度要求不高
                "  'lookup.cache.max-rows' = '10', " +
                "  'lookup.cache.ttl' = '1 hour', " +
                "  'table-name' = 'base_dic', " +
                "  'username' = 'root', " +
                "  'password' = '000000' " +
                ")";
//        查询mysql构建LookUp表（只能用process时间）
        tableEnv.executeSql(sql);

//        打印LookUp表
//        tableEnv.sqlQuery("select * from base_dic")
//                .execute()
//                .print();

//        构建事实表
        //1001,23.6,1324
        SingleOutputStreamOperator<WaterSensor> waterSensorDS1 = env.socketTextStream("hadoop3-01", 1111)
                .map(line -> {
                    String[] split = line.split(",");
                    return new WaterSensor(split[0], Double.parseDouble(split[1]), Long.parseLong(split[2]));
                });
        Table table = tableEnv.fromDataStream(waterSensorDS1
                , $("id")
                , $("vc")
                , $("ts")
//                额外加一个字段作为字段时间
                , $("pt").proctime()
        );
        tableEnv.createTemporaryView("t1",table);

//        lookup.cache没有开启时（默认关闭）每来一次数据就去mysql查询一次
//        使用事实表关联维表并打印结果
        tableEnv.sqlQuery("select    " +
                "    t1.id,   " +
                "    t1.vc,   " +
                "    dic.dic_name   " +
                "from t1 join base_dic FOR SYSTEM_TIME AS OF t1.pt as dic   " +
                "on t1.id=dic.dic_code")
                .execute()
                .print();

    }
}
